package com.demo.kafka.controller;

import com.alibaba.fastjson.JSON;
import com.demo.kafka.bean.Programmer;
import com.demo.kafka.constant.Topic;
import com.demo.kafka.producer.KafkaProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;

@Slf4j
@RestController
public class SendMsgController {
    @Autowired
    private KafkaProducer producer;
    @Autowired
    private KafkaTemplate kafkaTemplate;

    @RequestMapping({"/", "", "/index"})
    public String index() {
        return "this is a index...";
    }

    @GetMapping("sendSimple")
    public String sendSimple() {
        producer.sendMessage(Topic.SIMPLE, "hello spring boot kafka");
        return "success";
    }

    @GetMapping("sendBean")
    public String sendBean() {
        Programmer programmer = new Programmer("tonghai", 12, new Date());
        producer.sendMessage(Topic.BEAN, JSON.toJSON(programmer).toString());
        return "success";
    }

    @GetMapping("sendGroup")
    public String sendGroup() {
        for (int i = 0; i < 4; i++) {
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(Topic.GROUP, i % 4, "key", "hello group " + i);
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable ex) {
                    log.info("send group msg 发送时失败:{}", ex);
                }

                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    log.info("send group msg 发送失败:{}", result.toString());
                }
            });
        }
        return "success";
    }


}
